package com.sh.data.engine.domain.datadev.flink.job;

import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.FlinkJobStatusEnum;
import com.sh.data.engine.common.enumDefinition.FlinkNodeTypeEnum;
import com.sh.data.engine.common.util.FlinkUtil.FlinkCheckpointInfo;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkJobDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkSqlTaskDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkTaskDomain;
import com.sh.data.engine.domain.datadev.flink.model.param.StartFlinkTaskParam;
import com.sh.data.engine.domain.datadev.flink.service.FlinkJobService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkSqlTaskService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkTaskService;
import com.sh.data.engine.domain.datadev.flinkTaskAboutExexte.FlinkTaskApiService;
import com.sh.data.engine.domain.util.LogUtil;
import com.sh.data.engine.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * @author: zigui.zdf
 * @description:
 * @date: 2021/6/15 17:21
 */
@Component
@Slf4j
public class StopAndRunFlinkTask {

    @Autowired
    private FlinkSqlTaskService flinkSqlTaskService;

    @Autowired
    private FlinkTaskService flinkTaskService;

    @Autowired
    private FlinkJobService flinkJobService;

    @Autowired
    private FlinkTaskApiService flinkTaskApiService;

    @Value("${task.stopAndRunFlinkTask.enabled:true}")
    private boolean enabled;

    @XxlJob("FlinkTaskStopAndRunFlinkTask")
    public void start() {
        if (!enabled) {
            return;
        }

        // 遍历正在running的
        final List<FlinkJobDomain> runningJobList = flinkJobService.getRunningJobList();

        log.info("running job count : {}", runningJobList.size());

        for (FlinkJobDomain flinkJobDomain : runningJobList) {
            try {
                final Long versionId = flinkJobDomain.getNodeConfigId();
                final Long taskNodeId = flinkJobDomain.getNodeId();
                final Integer taskType = flinkJobDomain.getTaskType();
                final Integer execType = flinkJobDomain.getExecType();

                // 将调试的任务过滤掉
                if (Objects.equals(execType, 0)) {
                    continue;
                }

                // 判断任务是否需要停止,满足以下某个条件说明任务要停止
                // 1.调度停止时间到达 2.版本变了 3.取消发布了 4.版本被删了
                StopTypeEnum needStop = this.needStop(versionId, taskNodeId, taskType);

                if (null != needStop) {
                    this.stopTask(flinkJobDomain.getId(), taskNodeId, taskType, execType, needStop);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }

        // 查出所有到达开始时间点的任务,正在运行、停止的任务将被过滤;如果是失败的任务并且版本id不变，则也过滤掉
        List<Integer> filterStatusList =
            Lists.newArrayList(
                FlinkJobStatusEnum.RUNNING.getCode(), FlinkJobStatusEnum.STOPPED.getCode());

        // flink app
        this.startFlinkTask(filterStatusList);

        // flink sql
        this.startFlinkSqlTask(filterStatusList);
    }

    /**
     * 判断任务是否需要停止
     *
     * @param versionId
     * @param taskNodeId
     * @param taskType
     * @return
     */
    private StopTypeEnum needStop(Long versionId, Long taskNodeId, Integer taskType) {
        StopTypeEnum needStop = null;

        Date nextStopTime = null;
        Long activeId = null;
        Integer isPublished = -1;

        if (Objects.equals(taskType, FlinkNodeTypeEnum.TASK_SQL.getCode())) {
            final FlinkSqlTaskDomain sqlTaskDomain = flinkSqlTaskService.getById(versionId);
            final FlinkSqlTaskDomain activeSqlTaskDomain =
                flinkSqlTaskService.getActiveByNodeId(taskNodeId);

            if (null == sqlTaskDomain || null == activeSqlTaskDomain) {
                needStop = StopTypeEnum.VersionDeleted;
                return needStop;
            }

            nextStopTime = sqlTaskDomain.getNextStopTime();
            activeId = activeSqlTaskDomain.getId();
            isPublished = sqlTaskDomain.getIsPublished();
        } else if (Objects.equals(taskType, FlinkNodeTypeEnum.TASK_JAR.getCode())) {
            final FlinkTaskDomain taskDomain = flinkTaskService.getById(versionId);
            final FlinkTaskDomain activeTaskDomain = flinkTaskService.getActiveByNodeId(taskNodeId);

            if (null == taskDomain || null == activeTaskDomain) {
                needStop = StopTypeEnum.VersionDeleted;
                return needStop;
            }

            nextStopTime = taskDomain.getNextStopTime();
            activeId = activeTaskDomain.getId();
            isPublished = activeTaskDomain.getIsPublished();
        }

        if (nextStopTime != null && nextStopTime.before(new Date())) {
            needStop = StopTypeEnum.StopTimeIsUp;
        } else if (!Objects.equals(versionId, activeId)) {
            needStop = StopTypeEnum.VersionChanged;
        } else if (isPublished == 0) {
            needStop = StopTypeEnum.UnPublish;
        }

        return needStop;
    }

    /**
     * 任务停止类型
     */
    private enum StopTypeEnum {
        // 停止时间到了
        StopTimeIsUp,
        // 版本变了
        VersionChanged,
        // 未发布
        UnPublish,
        // 版本删了
        VersionDeleted
    }

    private void stopTask(
        Long jobId, Long taskNodeId, Integer taskType, Integer execType, StopTypeEnum stopType) {
        try {
            flinkTaskApiService.stopFlinkTask(jobId, FlinkJobStatusEnum.PAUSED.getCode(), true);

            boolean includeToday = true;
            if (stopType.equals(StopTypeEnum.StopTimeIsUp)) {
                includeToday = false;
            }

            if (Objects.equals(execType, 1) && !stopType.equals(StopTypeEnum.VersionDeleted)) {
                if (Objects.equals(taskType, FlinkNodeTypeEnum.TASK_SQL.getCode())) {
                    flinkSqlTaskService.reCalculateNextStartAndStopTime(taskNodeId, includeToday);
                } else if (Objects.equals(taskType, FlinkNodeTypeEnum.TASK_JAR.getCode())) {
                    flinkTaskService.reCalculateNextStartAndStopTime(taskNodeId, includeToday);
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 拉起flinksql任务
     *
     * @param filterStatusList
     */
    private void startFlinkSqlTask(List<Integer> filterStatusList) {
        final List<FlinkSqlTaskDomain> flinkSqlTaskList = flinkSqlTaskService.getAllRunnableTaskList();

        for (FlinkSqlTaskDomain flinkSqlTask : flinkSqlTaskList) {
            try {
                final FlinkJobDomain jobDomain = flinkJobService.getByNodeId(flinkSqlTask.getNodeId(), 1);
                if (jobDomain != null) {
                    if (filterStatusList.contains(jobDomain.getStatus())) {
                        continue;
                    }

                    if (Objects.equals(jobDomain.getStatus(), FlinkJobStatusEnum.FAILURE.getCode())
                        && Objects.equals(flinkSqlTask.getId(), jobDomain.getNodeConfigId())) {
                        continue;
                    }
                }

                try {
                    StartFlinkTaskParam param = new StartFlinkTaskParam();
                    param.setNodeId(flinkSqlTask.getNodeId());
                    param.setExecType(1);
                    if (null != jobDomain) {
                        param.setSavePoint(this.getLastSavepoint(jobDomain.getId()));
                    }
                    flinkTaskApiService.startFlinkTask(param);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    try {
                        String logFileName = String.format("%s_%s", flinkSqlTask.getNodeId(), 1);

                        LogUtil.writeFlinkLog(
                            flinkSqlTask.getNodeId(),
                            Lists.newArrayList(
                                String.format(
                                    "[%s] %s",
                                    DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"),
                                    ExceptionUtils.getFullStackTrace(e))));
                    } catch (IOException ex) {
                        log.error(ex.getMessage(), ex);
                    }
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private void startFlinkTask(List<Integer> filterStatusList) {
        final List<FlinkTaskDomain> flinkTaskList = flinkTaskService.getAllRunnableTaskList();
        for (FlinkTaskDomain domain : flinkTaskList) {
            try {
                final FlinkJobDomain jobDomain = flinkJobService.getByNodeId(domain.getNodeId(), 1);
                if (jobDomain != null) {
                    if (filterStatusList.contains(jobDomain.getStatus())) {
                        continue;
                    }

                    if (Objects.equals(jobDomain.getStatus(), FlinkJobStatusEnum.FAILURE.getCode())
                        && Objects.equals(domain.getId(), jobDomain.getNodeConfigId())) {
                        continue;
                    }
                }

                try {
                    StartFlinkTaskParam param = new StartFlinkTaskParam();
                    param.setNodeId(domain.getNodeId());
                    param.setExecType(1);
                    if (null != jobDomain) {
                        param.setSavePoint(this.getLastSavepoint(jobDomain.getId()));
                    }
                    flinkTaskApiService.startFlinkTask(param);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    try {
                        String logFileName = String.format("%s_%s", domain.getNodeId(), 1);

                        LogUtil.writeFlinkLog(
                            domain.getNodeId(),
                            Lists.newArrayList(
                                String.format(
                                    "[%s] %s",
                                    DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"),
                                    ExceptionUtils.getFullStackTrace(e))));
                    } catch (IOException ex) {
                        log.error(ex.getMessage(), ex);
                    }
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private String getLastSavepoint(Long jobId) {
        if (null == jobId) {
            return null;
        }

        final FlinkCheckpointInfo lastSavepoint = flinkTaskApiService.getLastSavepoint(jobId);
        if (null != lastSavepoint) {
            return lastSavepoint.getCheckpointPath();
        }
        return null;
    }
}
